package com.ibm.cps.spark.streaming.adapter;

import backtype.storm.task.IOutputCollector;
import backtype.storm.tuple.Values;
import com.ibm.cps.message.AbstractMessage;
import com.ibm.cps.message.MessageFactory;
import com.ibm.cps.message.RawDataMessage;
import com.ibm.util.TopologyStreamIds;
import com.ibm.util.exception.CPSException;

/**
 * Created by telekinesis on 5/4/15.
 */
public class EmbeddedMessageSpout extends AbstractSparkEmbeddedSpout {

    public EmbeddedMessageSpout(IOutputCollector collector) {
        super(collector);
    }

    @Override
    public void emitNewData(String streamName, AbstractMessage data) {
        RawDataMessage dataMessage = (RawDataMessage)data;
        collector.emit(TopologyStreamIds.RAW_DATA_STREAM, null, new Values(dataMessage, dataMessage.getTsKey()));
    }
}